/**
 * Copyright © 2018 eqxiu.com 北京中网易企秀科技有限公司  All rights reserved.
 */
package com.eqxiu.iom.msg.service;

import com.alibaba.fastjson.JSONObject;
import com.eqxiu.common.codec.CodecFactory;
import com.eqxiu.iom.Model.RoutingMsgModel;
import com.eqxiu.iom.msg.dto.BehaviorEvent;
import com.eqxiu.iom.msg.manager.DubboManager;
import com.eqxiu.user.api.dto.ScoreRuleDTO;
import com.eqxiu.user.utils.UserConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @description 用户行为消息发送到MQ
 *       com.eqxiu.iom.msg.service.MessageService
 * @author wangl
 * @date 2020-05-28
 * @modified by xxx
 */
@Service
public class MessageService implements RabbitTemplate.ReturnCallback, RabbitTemplate.ConfirmCallback {

    public static final Logger LOGGER = LoggerFactory.getLogger(MessageService.class);

    @Value("${queue.exchange.name.msgcenter}")
    private String msgCenterExchange;

    @Value("${queue.routing.key.msgcenter}")
    private String msgCenterRoutingKey;

    @Value("${queue.exchange.name.user.behavior}")
    private String userBehaviorExchange;

    @Value("${queue.routing.key.user.behavior}")
    private String userBehaviorRoutingKey;

    @Value("${userBehavior.score.rule.filter.enable}")
    private Boolean scoreRuleFilterEnable;

    @Autowired
    private DubboManager dubboManager;

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * @description 发送消息到消息中心MQ
     * @param event
     */
    public void sendMsgCenterMessage(BehaviorEvent event) {
        try {
            // springboot 2.x 更换 eqxiu-boot-xxx modified by wangl 2020-05-28
//			msgCenterMessageProvider.sendMessage(CodecFactory.getCodec().serialize(convertToRoutingMsgModel(event)), msgCenter);
            dubboManager.getMaxcenterRabbitTemplate().convertAndSend(msgCenterExchange, msgCenterRoutingKey, CodecFactory.getCodec().serialize(convertToRoutingMsgModel(event)));
        } catch (Exception e) {
            LOGGER.error("Send message: " + JSONObject.toJSONString(event) + ", exception: ", e);
        }
    }

    /**
     * @description 发送消息到用户行为MQ
     * @param event
     */
    public void sendUserBehaviorMessage(BehaviorEvent event) {
        // 判断event的用户行为是否设置了积分规则 如未设置：只发送消息中心MQ
        // 判断是否开启过滤用户行为绑定积分规则功能 true：开启过滤  false：未开启过滤
        if (scoreRuleFilterEnable) {
            if (!isBindScoreRule(event.getEventCode())) {
                LOGGER.warn("Unbound Score Rules for this User's Behavior [{}]. The Message [{}] disCard. ", event.getEventCode(), event.toString());
                // 20190221 修复用户行为不发消息的bug
                //				return;
            }
        }

        // 发送消息到用户行为MQ
        try {
            LOGGER.info("scoreRuleFilterEnable: " + scoreRuleFilterEnable + ", userBehaviorExchange: " + userBehaviorExchange + ", userBehaviorRoutingKey: "
                    + userBehaviorRoutingKey + ", event: {}", CodecFactory.getCodec().serialize(event));
            // springboot 2.x 更换 eqxiu-boot-xxx modified by wangl 2020-05-28
//			userBehaviorMessageProvider.sendMessage(CodecFactory.getCodec().serialize(event), userBehavior);
            dubboManager.getUserRabbitTemplate().convertAndSend(userBehaviorExchange, userBehaviorRoutingKey, CodecFactory.getCodec().serialize(event));
        } catch (Exception e) {
            LOGGER.error("Send message: " + JSONObject.toJSONString(event) + ", exception: ", e);
        }
    }

    /**
     * @description 行为触发消息体转为MsgCenter消息体
     * @param event 行为触发消息体
     * @return
     */
    private List<RoutingMsgModel> convertToRoutingMsgModel(BehaviorEvent event) {
        List<RoutingMsgModel> list = new ArrayList<>(1);
        RoutingMsgModel msgModel = new RoutingMsgModel();
        msgModel.setProducerCode(event.getProducerCode());
        msgModel.setUserid(event.getUserId());
        msgModel.setUserType(event.getUserType() + "");
        msgModel.setEventCode(event.getEventCode());
        msgModel.setSendTime(String.valueOf(event.getTimestamp()));
        msgModel.setLoginName(event.getLoginName());

        // 业务消息变量
        Map<String, Object> params = new HashMap<>();
        Map<String, Object> map = new HashMap<>();
        if (event.getLoginName() != null) {
            map.put("loginName", event.getLoginName());
        }
        if (event.getParams() != null && !event.getParams().isEmpty()) {
            map.putAll(event.getParams());
        }
        params.put("params", map);
        msgModel.setBody(params);

        if (!CollectionUtils.isEmpty(event.getExt())) {
            msgModel.setExt(event.getExt());
        }

        list.add(msgModel);
        return list;
    }

    /**
     * @description 判断用户行为是否绑定了积分规则
     * @param behaviorCode
     * @return
     */
    private Boolean isBindScoreRule(String behaviorCode) {
        try {
            // 先从缓存获取
            String key = UserConstants.CACHE_SOCRE_RULE_KEY_PREFIX + behaviorCode;
            Object object = redisTemplate.opsForValue().get(key);
            if (object != null) {
                ScoreRuleDTO dto = (ScoreRuleDTO) object;
                // 未绑定积分规则 缓存 new ScoreRuleDTO() 对象
                if (dto.getId() != null) {
                    return true;
                }
            }
        } catch (Exception e) {
            LOGGER.error("Query cache exception. BehaviorCode: {}. Exception: {}", behaviorCode, e.getMessage());
            // 缓存异常 快速失败 默认绑定
            return true;
        }

        try {
            // 如缓存不存在则通过服务获取
            ScoreRuleDTO scoreRuleDTO = getScoreRuleByBehaviorCode(behaviorCode);
            // 判断是否为空
            return scoreRuleDTO != null && scoreRuleDTO.getId() != null ? Boolean.TRUE : Boolean.FALSE;
        } catch (Exception e) {
            LOGGER.error("Judging whether user behavior code [ " + behaviorCode + " ] binds to score rule exception: {}", e);
            // 发生异常默认true
            return true;
        }
    }

    /**
     * @description 根据行为编码通过服务获取积分规则
     * @param behaviorCode 行为编码
     * @return
     */
    private ScoreRuleDTO getScoreRuleByBehaviorCode(String behaviorCode) {
        // dubbo服务 缓存不存在从db获取并同步到缓存（用户行为未绑定积分规则 缓存 new ScoreRuleDTO() 对象）
        return dubboManager.getUserMemberApi().getScoreRuleByBehaviorCode(behaviorCode);
    }

    /**
     * 此时发送消息有两种错误情况和两种成功
     * <p>
     * 1:消息未到交换器,可能是网络原因或者是交换器名写错等等  
     * 如果是这种情况的话 rabbitmq会调用confirm方法,ack参数为false
     * <p>
     * 2:消息到达交换器
     * 如果是这种情况的话 rabbitmq会调用confirm方法,ack参数为true
     * <p>
     * 3:消息未到达队列,可能是没有队列和你发送的交换器进行绑定,或者网络原因
     * 如果是这种情况的话 rabbitmq会调用 returnedMessage方法
     * <p>
     * 4:消息成功到达队列
     * 不会进入returnedMessage方法
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String s) {
        //消息确认
        if (ack) {
            LOGGER.info("SendMessage  >>>  消息到达交换器: {}", correlationData.getId());
        } else {
            LOGGER.error("SendMessage  >>>  消息未到达交换器: {}", correlationData.getId());
        }
    }

    @Override
    public void returnedMessage(Message message, int i, String s, String s1, String s2) {
        LOGGER.error("SendMessage  >>>  消息未到达队列: {}", message.toString());
    }

}

